package com.netty.cn.rpc.protoc;

import java.util.concurrent.TimeUnit;

import com.netty.cn.rpc.protoc.client.ProtocClientHandler;
import com.netty.cn.rpc.protoc.model.ClientModel.ChildInfo;
import com.netty.cn.rpc.protoc.model.ClientModel.ClientDataModel;
import com.netty.cn.rpc.protoc.model.ClientModel.Datalist;
import com.netty.cn.rpc.protoc.model.ServerModel;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;

/**
 * 
 * @author wangzonghui
 * @date 2021-07-19 16:36:36
 * @Description proto类型netty客户端
 */
public class RpcProtocClient {
	
	private Channel channel;
	private Bootstrap bootstrap;
	
	private  String host="localhost";
	private  int port=9090;
	
	public static void main(String[] args) throws InterruptedException {
		RpcProtocClient rpcProtocClient = new RpcProtocClient();
		rpcProtocClient.start(rpcProtocClient);
		
		//暂停一下，给连接留时间
		TimeUnit.SECONDS.sleep(1);
		rpcProtocClient.sendData();
	}
	
	private void sendData() throws InterruptedException {
		if (channel != null && channel.isActive()) {
			
			for(int i=0;i<3;i++) {
				ClientDataModel.Builder clientHeart=ClientDataModel.newBuilder();
				clientHeart.setMsg("客户端消息");
				
				ChildInfo.Builder childInfo = ChildInfo.newBuilder();
				childInfo.setContext("子类内容");
				
				Datalist.Builder datalist =Datalist.newBuilder();
				
				datalist.setValue(10);
				
				clientHeart.setChildInfo(childInfo);
				clientHeart.addDatalist(datalist);
				
				
			    channel.writeAndFlush(clientHeart.build());
			    TimeUnit.SECONDS.sleep(2);
			}
			
		}

		
	}

	/**
	 * 启动服务
	 * 
	 * @throws InterruptedException
	 */
	public void start(RpcProtocClient rpcProtocClient) {

		EventLoopGroup group = new NioEventLoopGroup();
		bootstrap = new Bootstrap();
		bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {

			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ChannelPipeline channelPipeline = ch.pipeline();
            	//解码
            	channelPipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder())
                .addLast("protobufDecoder",new ProtobufDecoder(ServerModel.ServerDataModel.getDefaultInstance()))  
                //编码规则
                .addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender())
                .addLast("protobufEncoder", new ProtobufEncoder())  
                // ----Protobuf处理器END----  
                .addLast(new ProtocClientHandler(rpcProtocClient));
			}
		});

		doConnect();
	}
	
	/**
	 * 负责客户端和服务器的 TCP 连接的建立, 并且当 TCP 连接失败时, doConnect 会 通过
	 * "channel().eventLoop().schedule" 来延时10s 后尝试重新连接
	 */
	public void doConnect() {
		if (channel != null && channel.isActive()) {
			return;
		}

		ChannelFuture future = bootstrap.connect(host, port);

		future.addListener(new ChannelFutureListener() {
			public void operationComplete(ChannelFuture futureListener) throws Exception {
				if (futureListener.isSuccess()) {
					channel = futureListener.channel();
					System.out.println("连接服务端成功!");
				} else {
					System.out.println("连接失败 ，10s 后尝试连接 "+host+":"+port);

					futureListener.channel().eventLoop().schedule(new Runnable() {
						@Override
						public void run() {
							doConnect();
						}
					}, 10, TimeUnit.SECONDS);
				}
			}
		});
	}

}
